package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}

object Demo12PageRank {
  def main(args: Array[String]): Unit = {
    //1、创建spark配置文件对象
    val conf = new SparkConf()
    // 运行方式
    conf.setMaster("local") //本地运行
    conf.setAppName("WordCount") //程序名

    val q = 0.85 //阻尼系数

    //2、创建spark上下文对象
    val sc = new SparkContext(conf)

    val prRDD = sc
      .textFile("data/pagerank.txt")
      .map(line => {
        val split = line.split("\t")
        val page = split(0)
        val link = split(1).split(",").toList //出链列表

        (page, link)
      })

    //计算网页数量
    val N = prRDD.count()

    //初始pr值
    var pr = prRDD.mapValues(x => 1.0)

    var flag = true

    //循环迭代
    while (flag) {
      var newPR = pr.join(prRDD).flatMap(kv => {
        val page = kv._1
        val pr = kv._2._1
        val link = kv._2._2

        //将网页pr分配给他的出链网页
        link.map(p => (p, pr / link.length))
      }).reduceByKey(_ + _) //计算新的pr值

      //增加阻尼系数
      newPR = newPR.mapValues(pr => (1 - q) / N + q * pr)

      //设定一个差值指标（0.0001）。当所有页面和上一次计算的PR差值平均小于该标准时，则收敛。
      val sumPR = newPR
        .join(pr)
        .mapValues(pr => Math.abs(pr._1 - pr._2))
        .map(_._2)
        .sum()

      //差值平均值
      val i = sumPR / N


      //实现迭代
      pr = newPR


      //收敛
      if (i < 0.001) {
        flag = false
      }
    }


    pr.foreach(println)

  }
}
